package main.scala.com.hu.wc

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * @Author: hujianjun
 * @Create Date: 2020/12/2 9:39
 * @Describe: 按省份统计广告点击量
 */
//定义广告日志和输出结果样例类
case class ODPSInfo(dateId: String, accountId: String, premium: Double, total: Double)

object AdClickStatistic {
  def main(args: Array[String]): Unit = {
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度为1，时间语义为事件时间
    streamEnv.setParallelism(1)
//    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //读取文件数据源
    val inputStream = streamEnv.readTextFile("D:\\hjj\\T\\test-pdf\\odps.csv")

    //转换数据流为样例类流
    val dataStream = inputStream.map(data => {
      val arr = data.split(",")
      ODPSInfo(arr(0),  arr(1), arr(2).toDouble, arr(3).toDouble)
    })

    //根据省份分区统计广告点击量
    val resultStream = dataStream.keyBy(_.accountId)
      .timeWindow(Time.days(3), Time.days(1)) //开滑动窗口，每5s统计过去1小时的省份的点击量
      .aggregate(new PreAggregateADClickCnt2(), new ADClickCntResultWindow2()) //使用aggregate函数进行统计,第一个自定义函数实现预聚合，第二个输出带有窗口时间结果

    resultStream.print()

    streamEnv.execute("ad odps statistic job")
  }
}

class PreAggregateADClickCnt2 extends AggregateFunction[ODPSInfo, Double, Double] {
  override def createAccumulator(): Double = 0.0

  override def add(in: ODPSInfo, acc: Double): Double = acc + in.premium

  override def getResult(acc: Double): Double = acc

  override def merge(acc: Double, acc1: Double): Double = acc + acc1
}

class ADClickCntResultWindow2 extends WindowFunction[Double, ODPSInfo, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Double], out: Collector[ODPSInfo]): Unit = {
    out.collect(ODPSInfo(key, window.getEnd.toString, input.last,0.0))
  }
}